package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.UserLoginDwd;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class DwdUserUserLogin extends BaseApp {
    public static void main(String[] args) {
        new DwdUserUserLogin().start(
                11103,
                4,
                "dwd_user_user_login",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print("page");
        //转换数据，将jsoNStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        try {
                            if (jsonObject.getJSONObject("common").getString("uid") != null) {
                                return jsonObject;
                            }
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
                        return null;
                    }
                }
        );
        //提取事件时间和添加水位线
        SingleOutputStreamOperator<JSONObject> withTimestampsAndWatermarks = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                (element, recordTimestamp) -> element.getLong("ts")
                        )
        );
        //按照mid进行分组
        KeyedStream<JSONObject, String> keyByDS =
                withTimestampsAndWatermarks.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //使用flink状态编程找出每个会话的第一条数据
        SingleOutputStreamOperator<JSONObject> firstLoginDS = keyByDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> firstLoginDateValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<>("firstLoginDateValueState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                        firstLoginDateValueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject firstLoginJsonObj = firstLoginDateValueState.value();
                        Long firstLoginTs = jsonObject.getLong("ts");
                        if (firstLoginJsonObj==null){
                            firstLoginDateValueState.update(jsonObject);
                            ctx.timerService().registerProcessingTimeTimer(firstLoginTs+10*1000L);
                        }else {
                            Long lastTs = firstLoginJsonObj.getLong("ts");
                            if (firstLoginTs<lastTs){
                                firstLoginDateValueState.update(jsonObject);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        out.collect(firstLoginDateValueState.value());
                    }
                }
        );

        //将数据进行转换，jsonObj->UserLoginDwd->jsonStr
        SingleOutputStreamOperator<String> mapDS = firstLoginDS.map(
                new MapFunction<JSONObject, String>() {
                    @Override
                    public String map(JSONObject jsonObj) throws Exception {
                        JSONObject common = jsonObj.getJSONObject("common");
                        Long ts = jsonObj.getLong("ts");
                        String loginTime = DateFormatUtil.tsToDateTime(ts);
                        String dateId = loginTime.substring(0, 10);

                        UserLoginDwd userLoginDwd = UserLoginDwd.builder()
                                .userId(common.getString("uid"))
                                .dateId(dateId)
                                .loginTime(loginTime)
                                .channel(common.getString("ch"))
                                .provinceId(common.getString("ar"))
                                .versionCode(common.getString("vc"))
                                .midId(common.getString("mid"))
                                .brand(common.getString("ba"))
                                .model(common.getString("md"))
                                .sourceId(common.getString("sc"))
                                .operatingSystem(common.getString("os"))
                                .ts(ts)
                                .build();

                        return JSON.toJSONString(userLoginDwd);
                    }
                }
        );
        //{"brand":"Honor","channel":"xiaomi","dateId":"2024-07-06",
        // "loginTime":"2024-07-06 15:38:27","midId":"mid_305","model":"Honor 20s",
        // "operatingSystem":"Android 10.0","provinceId":"31","sourceId":"1",
        // "ts":1720251507856,"userId":"857","versionCode":"v2.1.134"}
        mapDS.print("mapDS");
        //输出到kafka
        mapDS.sinkTo(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_USER_LOGIN));

    }
}
